package com.we.risk.registerchanneldeprecated.browseinfo.dim;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.util.Date;

public class TableBrowseOriginData {
    public static final String KFKTABLE = "bdm.kafka_browse_origin_data_inc";
    public static final String DATEFORMAT = "yyyy-MM-dd";
    public static final String EARLESTTIME = "earliest_recv_time";
    public static final String LATESTTIME = "latest_recv_time";

    public static Logger LOG = LoggerFactory.getLogger(TableBrowseOriginData.class);

    // dt>2023-01-15 00:00:00
    public static final String QUERYSQL =
            " select unique_channel, receive_time, idfa, oaid "
                    + "from "
                    + "("
                    + " select cast(IFNULL(JSON_VALUE(raw_data, '$.receive_time'), '0') as bigint)/1000 as receive_time,"
                    + " JSON_VALUE(raw_data, '$.idfa') as idfa,"
                    + " JSON_VALUE(raw_data, '$.oaid') as oaid,"
                    + " JSON_VALUE(raw_data, '$.unique_channel') as unique_channel"
                    + "   from "
                    + KFKTABLE
                    + ")"
                    + "where receive_time >= 1673712000"
                    + " and receive_time <> 0";
    public String unique_channel;
    public long receive_time;
    public String idfa;
    public String oaid;

    public TableBrowseOriginData() {}

    public TableBrowseOriginData(
            String unique_channel, long receive_time, String idfa, String oaid) {
        this.unique_channel = unique_channel;
        this.receive_time = receive_time;
        this.idfa = idfa;
        this.oaid = oaid;
    }

    public static TableBrowseOriginData init() {
        return new TableBrowseOriginData("", -1, "", "");
    }

    public static SingleOutputStreamOperator<String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        tableEnv.useCatalog("cdh6_hive");
        Table tableRes = tableEnv.sqlQuery(TableBrowseOriginData.QUERYSQL);
        DataStream<TableBrowseOriginData> inputStream =
                tableEnv.toDataStream(tableRes, TableBrowseOriginData.class);

        SingleOutputStreamOperator<String> out =
                inputStream
                        .flatMap(new TableBrowseOriginDataCleanFunc())
                        .keyBy(
                                new KeySelector<BrowseOriginInfo, String>() {
                                    @Override
                                    public String getKey(BrowseOriginInfo value) throws Exception {
                                        return value.getId_combine();
                                    }
                                })
                        .flatMap(new TableBrowseOriginDataProcess());

        return out;
    }

    public static class TableBrowseOriginDataProcess
            extends RichFlatMapFunction<BrowseOriginInfo, String> {
        MapStateDescriptor<Long, JSONObject> mStateDesc;
        MapState<Long, JSONObject> browseOrgMState;
        SimpleDateFormat sdf;

        @Override
        public void open(Configuration parameters) throws Exception {
            StateTtlConfig ttlConfig =
                    StateTtlConfig.newBuilder(Time.hours(24)) // dt 按天分区 只存24小时
                            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                            .cleanupInRocksdbCompactFilter(100000L)
                            .build();

            mStateDesc = new MapStateDescriptor<>("browseOrgMState", Long.class, JSONObject.class);
            mStateDesc.enableTimeToLive(ttlConfig);

            browseOrgMState = getRuntimeContext().getMapState(mStateDesc);

            sdf = new SimpleDateFormat(DATEFORMAT);
        }

        @Override
        public void flatMap(BrowseOriginInfo value, Collector<String> out) throws Exception {
            try {
                String id_combine = value.getId_combine();
                String unique_channel = value.getUnique_channel();
                long receive_time = value.getReceive_time();

                // dt_time 秒级时间戳
                long dt_time = getStartTimeOfDay(receive_time * 1000, sdf);

                JSONObject outJson = new JSONObject();
                outJson.put("id_combine", id_combine);
                outJson.put("unique_channel", unique_channel);
                outJson.put("dt_time", dt_time);

                // 分区判断
                // LOG.warn("contains dt_time: " + dt_time + " " +
                // browseOrgMState.contains(dt_time));
                if (browseOrgMState.contains(dt_time)) {
                    JSONObject dtJson = browseOrgMState.get(dt_time);

                    if (dtJson != null) {
                        // channel 判断
                        // LOG.warn("contains: " + unique_channel + " " +
                        // dtJson.containsKey(unique_channel + EARLESTTIME));
                        if (dtJson.containsKey(unique_channel + EARLESTTIME)) {
                            long earliest_time = dtJson.getLongValue(unique_channel + EARLESTTIME);
                            long latest_time = dtJson.getLongValue(unique_channel + LATESTTIME);

                            if (receive_time < earliest_time) {
                                dtJson.put(unique_channel + EARLESTTIME, receive_time);
                                outJson.put("label_name", EARLESTTIME);
                                outJson.put("label_value", receive_time);
                                out.collect(outJson.toString());
                                // LOG.warn("old channel outJson1: " + outJson.toString());
                                browseOrgMState.put(dt_time, dtJson);
                            }

                            if (receive_time > latest_time) {
                                dtJson.put(unique_channel + LATESTTIME, receive_time);
                                outJson.put("label_name", LATESTTIME);
                                outJson.put("label_value", receive_time);
                                out.collect(outJson.toString());
                                // LOG.warn("old channel outJson2: " + outJson.toString());
                                browseOrgMState.put(dt_time, dtJson);
                            }
                        } else {
                            // LOG.warn("##new channel");
                            dtJson.put(unique_channel + EARLESTTIME, receive_time);
                            outJson.put("label_name", EARLESTTIME);
                            outJson.put("label_value", receive_time);
                            // LOG.warn("new channel outJson1: " + outJson.toString());
                            out.collect(outJson.toString());

                            dtJson.put(unique_channel + LATESTTIME, receive_time);
                            outJson.put("label_name", LATESTTIME);
                            outJson.put("label_value", receive_time);
                            // LOG.warn("new channel outJson2: " + outJson.toString());
                            out.collect(outJson.toString());

                            browseOrgMState.put(dt_time, dtJson);
                        }
                    }
                } else {
                    // LOG.warn("##new dt ");
                    JSONObject dtJson = new JSONObject();
                    dtJson.put(unique_channel + EARLESTTIME, receive_time);
                    outJson.put("label_name", EARLESTTIME);
                    outJson.put("label_value", receive_time);
                    // LOG.warn("new dt outJson1: " + outJson.toString());
                    out.collect(outJson.toString());

                    dtJson.put(unique_channel + LATESTTIME, receive_time);
                    outJson.put("label_name", LATESTTIME);
                    outJson.put("label_value", receive_time);
                    // LOG.warn("new dt outJson2: " + outJson.toString());
                    out.collect(outJson.toString());

                    browseOrgMState.put(dt_time, dtJson);
                }
            } catch (Exception e) {
                LOG.error("excep: " + e.toString());
                LOG.error("excep input: " + value);
            }
        }
    }

    public static class TableBrowseOriginDataCleanFunc
            extends RichFlatMapFunction<TableBrowseOriginData, BrowseOriginInfo> {
        @Override
        public void flatMap(TableBrowseOriginData value, Collector<BrowseOriginInfo> out)
                throws Exception {
            String idfa = value.getIdfa();
            String oaid = value.getOaid();
            String id_combine = "";
            long receive_time = value.getReceive_time();
            String unique_channel = value.getUnique_channel();

            if (idfa != null && !("").equals(idfa) && !("__IDFA__").equals(idfa)) {
                id_combine = idfa;
            } else if (oaid != null && !("").equals(oaid) && !("__OAID__").equals(oaid)) {
                id_combine = oaid;
            }

            if ("00000000-0000-0000-0000-000000000000".equals(id_combine)
                    || ("").equals(id_combine)) {
                return;
            }

            BrowseOriginInfo browseOriginInfo =
                    new BrowseOriginInfo(id_combine, unique_channel, receive_time);
            out.collect(browseOriginInfo);
        }
    }

    private static long getStartTimeOfDay(long now, SimpleDateFormat sdf) {

        String date = sdf.format(new Date(now));
        long res = 0;
        try {
            res = sdf.parse(date).getTime() / 1000;
        } catch (Exception e) {
            LOG.error("sdf parse date error! time: " + now);
            e.printStackTrace();
        }
        return res;
    }

    public String getUnique_channel() {
        return unique_channel;
    }

    public long getReceive_time() {
        return receive_time;
    }

    public String getIdfa() {
        return idfa;
    }

    public String getOaid() {
        return oaid;
    }
}
